package com.xdf.testkafk.rocket;

import com.alibaba.fastjson.JSON;
import com.xdf.testkafk.vo.KafkaMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

/**
 * @author chanchaw
 * @create 2025-11-01 8:00
 */
@RequiredArgsConstructor
@Slf4j
@Component
public class RocketMQProducer {
    private static final String TOPIC = "imooc-study-rocketmq";
    private final RocketMQTemplate rocketMQTemplate;

    /**
     * 使用同步的方式发送消息，不指定 key 和 tag
     * 发送的数据是字符串类型
     * @param value
     */
    public void sendMessageWithValue(String value){
        SendResult sendResult = rocketMQTemplate.syncSend(TOPIC, value);
        log.info("rocketMQ同步的方式发送消息：[{}]", JSON.toJSONString(sendResult));

        SendResult sendResultOrderly = rocketMQTemplate.syncSendOrderly(TOPIC, value, "chanchaw");
        log.info("sendResultOrderly的方式发送消息：[{}]", JSON.toJSONString(sendResultOrderly));
    }

    /**
     * 使用异步的方式发送消息，指定 key
     * @param key
     * @param value
     */
    public void sendMessageWithKey(String key, String value){
        Message<String> message = MessageBuilder.withPayload(value).setHeader(RocketMQHeaders.KEYS, key).build();
        // 异步发送消息，需要回调函数
        rocketMQTemplate.asyncSend(TOPIC, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("RocketMQ发送消息成功:[{}]", JSON.toJSONString(sendResult));
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("RocketMQ异步方式发送消息出现异常：[{}]", throwable.getMessage(), throwable);
            }
        });
    }

    /**
     * 使用同步的方式发送消息，带有 tag，发送的消息是 java pojo
     * @param tag
     * @param value
     */
    public void sendMessageWithTag(String tag, String value){
        KafkaMessage msg = JSON.parseObject(value, KafkaMessage.class);
        SendResult sendResult = rocketMQTemplate.syncSend(TOPIC + ":" + tag, msg);
        log.info("同步的方式发送Java pojo：[{}]", JSON.toJSONString(sendResult));
    }

    /**
     * 同步的方式发送消息，带有 key 和 tag
     * @param key
     * @param tag
     * @param value
     */
    public void sendMessageWithAll(String key, String tag, String value){
        Message<String> message = MessageBuilder.withPayload(value).setHeader(RocketMQHeaders.KEYS, key).build();
        SendResult sendResult = rocketMQTemplate.syncSend(TOPIC + ":" + tag, message);
        log.info("同步的方式发送消息带有key 和 tag:[{}]", JSON.toJSONString(sendResult));
    }
}
